#ifndef MQSERVER_MQ_QUEUE_HPP_
#define MQSERVER_MQ_QUEUE_HPP_
#include "../mqcomm/mq_logger.hpp"
#include "../mqcomm/mq_helper.hpp"
#include <iostream>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <cassert>

namespace mqserver
{
class MsgQueue;
class MsgQueueManager;
using MsgQueuePtr = std::shared_ptr<MsgQueue>;
using MsgQueuePtrMap = std::unordered_map<std::string, MsgQueuePtr>;
using MsgQueueManagerPtr = std::shared_ptr<MsgQueueManager>;

class MsgQueue
{
public:
    MsgQueue() = default;
    MsgQueue(const std::string name, bool durable,
            bool exclusive, bool auto_delete,
            const google::protobuf::Map<std::string, std::string>& args_)
        : name_(name), durable_(durable), exclusive_(exclusive), auto_delete_(auto_delete), args_(args_)
    {}
    std::string getArgs()
    {
        return std::move(helper::StringHelper::serializeToString(args_));
    }
    // 反序列化args_str
    void setArgs(const std::string& args_str)
    {
        helper::StringHelper::deserializeFromString(args_str, &args_);       
    }
public:
    std::string name_;
    bool durable_;
    bool exclusive_;
    bool auto_delete_;
    google::protobuf::Map<std::string, std::string> args_;
};

class MsgQueueMapper
{
public:
    MsgQueueMapper(const std::string& name) : sql_helper_(name) 
    {
        std::string path = helper::FileHelper::parentDirectory(name);
        helper::FileHelper::createDirectory(path);
        assert(sql_helper_.open());
        createTable();
    }
    ~MsgQueueMapper() { sql_helper_.close(); }
    
    void createTable()
    {
        const std::string create_table_sql =
            "create table if not exists queue_table( \
            name varchar(32) primary key, \
            durable int, \
            exclusive int, \
            auto_delete int, \
            args varchar(128));";

        bool ret = sql_helper_.exec(create_table_sql, nullptr, nullptr);
        if(ret == false)
        {
            LOG_FATAL << "create queue_table fail!\n";
            abort();
        }
    }
    void removeTable()
    {
        const std::string drop_table_sql = "drop table if exists queue_table;";
        bool ret = sql_helper_.exec(drop_table_sql, nullptr, nullptr);
        if(ret == false) 
        {
            LOG_FATAL << "drop queue_table fail!\n";
            abort();
        }
    }
    bool insert(MsgQueuePtr& msg_queue)
    {
        const std::string insert =
            "insert into queue_table values('%s', %d, %d, %d, '%s')";
        std::string args_str = msg_queue->getArgs();
        char insert_sql[4096] = {0};
        sprintf(insert_sql, insert.c_str(), msg_queue->name_.c_str(), msg_queue->durable_, 
                msg_queue->exclusive_, msg_queue->auto_delete_, args_str.c_str());
        return sql_helper_.exec(insert_sql, nullptr, nullptr);
    }
    bool remove(const std::string& name)
    {
        const std::string delete_sql = "delete from queue_table \
                                        where name = '" + name + "'";
        return sql_helper_.exec(delete_sql, nullptr, nullptr);
    }
    MsgQueuePtrMap recovery()
    {
        MsgQueuePtrMap result;
        const std::string select_sql = "select name, durable, exclusive, auto_delete, args \
                                        from queue_table;";
        sql_helper_.exec(select_sql, selectCallback, (void*)&result);
        return result;
    } 
private:
    static int selectCallback(void* arg, int col_count, char** row, char** fields_name)
    {
        MsgQueuePtrMap* result = static_cast<MsgQueuePtrMap*>(arg);
        auto mqp = std::make_shared<MsgQueue>();
        mqp->name_ = row[0];
        mqp->durable_ = static_cast<bool>(std::stoi(row[1]));
        mqp->exclusive_ = static_cast<bool>(std::stoi(row[2]));
        mqp->auto_delete_ = static_cast<bool>(std::stoi(row[3]));
        if(row[4]) mqp->setArgs(row[4]);

        result->emplace(mqp->name_, mqp);
        return 0;        // 返回0才代表成功
    }

private:
    helper::SqliteHelper sql_helper_;
};

class MsgQueueManager
{
public:
    MsgQueueManager(const std::string& dbfile) 
        : mapper_(dbfile) { msg_queues_ =  mapper_.recovery(); }

    bool declareQueue(const std::string name, bool durable, 
            bool exclusive, bool auto_delete,
            const google::protobuf::Map<std::string, std::string>& args_)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        if(existsNoLock(name)) return true;
        MsgQueuePtr mqptr = std::make_shared<MsgQueue>(name, durable, exclusive, auto_delete, args_);
        if(durable == true)
        {
            int ret = mapper_.insert(mqptr);
            if(!ret) return false;
        }
        msg_queues_.emplace(name, mqptr);
        return true;
    }

    void deleteQueue(const std::string& name)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        if(!existsNoLock(name)) return;
        if(msg_queues_[name]->durable_ == true) mapper_.remove(name);
        msg_queues_.erase(name);
    }

    MsgQueuePtr selectQueue(const std::string& name)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        auto it = msg_queues_.find(name);
        return it != msg_queues_.end() ? it->second : MsgQueuePtr();
    }

    MsgQueuePtrMap getAllQueues()
    {
        std::lock_guard<std::mutex> lock(mtx_);
        return msg_queues_;
    }

    bool exists(const std::string& name)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        auto it = msg_queues_.find(name);
        return it != msg_queues_.end();
    }

    size_t size()
    {
        std::lock_guard<std::mutex> lock(mtx_);
        return msg_queues_.size();
    }

    void clear()
    {
        std::lock_guard<std::mutex> lock(mtx_);
        mapper_.removeTable();
        msg_queues_.clear();
    }
private:
    bool existsNoLock(const std::string& name)
    {
        auto it = msg_queues_.find(name);
        return it != msg_queues_.end();
    }
private:
    std::mutex mtx_;
    MsgQueueMapper mapper_;
    MsgQueuePtrMap msg_queues_;
};

} // namespace mqserver

#endif